package com.atguigu.sparkstreaming.examples

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/7/15
 *
 *
 *
 *    解决：   ①取消自动提交offset
 *              改为手动提交
 *            ②在输出成功后，再手动提交
 *
 *    ------------------------------
 *      写的代码，哪些在Driver端运行？ 哪些在Executor端运行?
 *
 *    ---------------------------------
 * Exception in thread "main" java.lang.IllegalArgumentException:
 *    requirement failed:
 *        No output operations registered, so nothing to execute
 *
 *
 *    ----------------------------------
 *    OffsetRange(topic: 'topicA', partition: 2, range: [62 -> 62])
 *        一个OffsetRange代表消费的一个分区一个主体，当前批次拉取到数据的起始offset和终止offset
 *            关注until offset
 *
 *    ------------------------------------
 *        local[*]: 本地模式以多线程模拟分布式运算。
 *              Executor的线程统一命名为 : Executor task launch worker for task id
 *              只要不是以上的线程名，都是Driver端。
 *
 *     ---------------------------
 *      如何区分代码在Driver端还是Executor?
 *          ①如果是DStream的普通(RDD中也有的，同名)算子例如:  map,filter等，都是在Executor端运行
 *          ②特殊算子
 *                transform 和  foreachRDD
 *                      只有 RDD.算子(xxx ) xxx是在Executor端运行，其他的位置都是在Driver端！
 *
 *     ----------------------------
 *      结论：  偏移量offsets是在Driver端获取！
 *             在Driver端提交!
 *
 *
 */
object AtLeastOnceDemo {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "TransformDemo", Seconds(10))

    //所有的消费者参数都可以在 ConsumerConfig
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "2203092",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "false"
    )


    val topics = Array("topicA")

    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )


    //在Driver端
    var ranges: Array[OffsetRange] = null
    //获取偏移量
    val ds1: DStream[ConsumerRecord[String, String]] = ds.transform(rdd => {

      //  Driver端
      println("aaaa:"+Thread.currentThread().getName)
      //偏移量
       ranges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

       ranges.foreach(println(_))

      rdd.map(record => {
        println("cccc:"+Thread.currentThread().getName)
        record
      })

    })

    ds1.map(record => {
      Thread.sleep(500)
     /* if(record.value().equals("B")){
        throw new RuntimeException("程序异常");
      }*/
      record.value()
    }).foreachRDD(rdd => {

      //输出  在Executor端
      rdd.foreach(word => println(Thread.currentThread().getName + ":"+word))

      //在Driver端
      println("bbbb:"+Thread.currentThread().getName)
      //输出之后，提交偏移量
      //手动提交offsets
      // 在Driver端运行
      ds.asInstanceOf[CanCommitOffsets].commitAsync(ranges)

    })

    // 不能直接暴露在Driver的main方法中，而应该在输出Operation中输出之后，调用
    // ds.asInstanceOf[CanCommitOffsets].commitAsync(ranges)


    // 启动APP
    streamingContext.start()

    // 阻塞进程，让进程一直运行
    streamingContext.awaitTermination()

  }

}
